package com.saic.data.dotask;

import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;

import com.saic.data.table.TableAction;
import com.saic.data.table.TableFile;
import com.saic.data.table.TableUtil;
import com.saic.data.util.DBUtil;
import com.saic.data.util.HbaseUtil;

public class UpdataDB extends DoTask{
	private TableFile tfile = null;
	private DBUtil sqlUtil = null;
	private String mode  = null;

	public UpdataDB(TableFile tfile, DBUtil sqlUtil) {
		this.tfile = tfile;
		this.sqlUtil = sqlUtil;
		this.mode = "new";
	}
	
	public UpdataDB(TableFile tfile, DBUtil sqlUtil,String mode) {
		this.tfile = tfile;
		this.sqlUtil = sqlUtil;
		this.mode = mode;
	}

	@Override
	public void doTask() {
		boolean flag = true;
		ResultScanner rs = null;
		try {
			rs = HbaseUtil.QueryByConditionValue(tfile.gettConf().getTmpHtbl(),tfile.gettConf().dataFamily,
					"fileName",tfile.getFileName());
			TableUtil.pm(this.getClass(), "Search "+tfile.gettConf().getTmpHtbl()+" by fileName= " +tfile.getFileName() );
			/** Delete ORACLE */
			if(tfile.gettConf().DBdelflag){
				flag = sqlUtil.deleteByMonth(tfile.gettConf().getOracleTbl(),tfile.getFinanceYear(),tfile.getFinanceMonth());
			    if(!flag) TableUtil.pm(this.getClass(),"SQL ERROR WHEN DELETE RECORD!");
			}
			int line = 0;
			int error = 0;
			for (Result r : rs) {

				/** Generate ORACLE insert CODE */
				String sql = tfile.getInsertSQL(r);
				if (sql.isEmpty()) continue;
				
				
				/** write to ORACLE DB */
				flag = sqlUtil.update(sql);
				if (!flag) {
					TableUtil.pm(this.getClass(),"SQL ERROR: "+sql);
					error++;
					continue;
				}

			
				/** Insert into csv_done */
				List<String> doneCol = tfile.gettConf().getColList();
				doneCol.add("UUID");
				doneCol.add("fileKey");
				doneCol.add("fileName");
				doneCol.add("upLoadDate");
				doneCol.add("streamNo");
				doneCol.add("recordDate");
				Put put= HbaseUtil.getPutFromResult(		
						tfile.gettConf().getHistoryHtbl(), r,tfile.gettConf().dataFamily,doneCol);
				flag = HbaseUtil.insertData(tfile.gettConf().getHistoryHtbl(),put);
				//TableUtil.pm(this.getClass(),"HbaseUtil.insertData "+tfile.gettConf().getHistoryHtbl()+" : "+ flag);

				/** Delete from csv_tmp */
				flag=HbaseUtil.deleteRow(tfile.gettConf().getTmpHtbl(),	r.getRow());
				//TableUtil.pm(this.getClass(),"HbaseUtil.deleteRow "+tfile.gettConf().getTmpHtbl()+" : "+ flag);
				line++;
			}
            
			if (line==0) {
				TableUtil.pm("UpdataDB-doTask", tfile.getFileName()+ ": no line for Update .Please Check Data");
				return;
			}
			
			/** Update FTP file DB */
			long tm = System.currentTimeMillis();
			TableUtil.pm(this.getClass(), "Table :"+tfile.getFileName()+" Update Lines : " + line);
			
			if(mode.compareTo("new")==0)
				HbaseUtil.updateData(TableUtil.configtab, tfile.getRowkey().getBytes(),
					TableUtil.configCustomFamily,"oracleRecordLines", line+"" );
			
			if(mode.compareTo("continus")==0){
				Result result = HbaseUtil.QueryByConditionKey(TableUtil.configtab, tfile.getRowkey());
				String lastline = HbaseUtil.getFromResult(result, TableUtil.configCustomFamily, "oracleRecordLines");
				line = line + Integer.parseInt(lastline);
				HbaseUtil.updateData(TableUtil.configtab, tfile.getRowkey().getBytes(),
						TableUtil.configCustomFamily,"oracleRecordLines", line+"" );
			}
			HbaseUtil.updateData(TableUtil.configtab, tfile.getRowkey().getBytes(),
					TableUtil.configCustomFamily,"oracleRecordTimes", tm + "");
			
			/** Update FTP file Stats */
			if(error ==0){
				HbaseUtil.updateData(TableUtil.configtab, tfile.getRowkey().getBytes(),
					TableUtil.configtabFamily,"fileStats", TableAction.Done.toString());}
			else{
				HbaseUtil.updateData(TableUtil.configtab, tfile.getRowkey().getBytes(),
					TableUtil.configtabFamily,"fileStats", TableAction.ErrorStop.toString());
				
			}
			
			/** Update month_streamNo */
			if(tfile.gettConf().DBdelflag && mode.compareTo("new")==0){
				String streamKey = tfile.gettConf().tablename +tfile.getFinanceYear()+ tfile.getFinanceMonth();
				Put put = new Put(streamKey.getBytes());
				put.add(TableUtil.streamcheckfamily.getBytes(),"streamNo".getBytes(),tfile.getStreamNo().getBytes());
				put.add(TableUtil.streamcheckfamily.getBytes(),"modifieddate".getBytes(),tfile.getModifieddate().getBytes());
				put.add(TableUtil.streamcheckfamily.getBytes(),"fileRowkey".getBytes(),tfile.getRowkey().getBytes());
				HbaseUtil.insertData(TableUtil.streamchecktab,put);
			}

		} catch (Exception e) {
			TableUtil.pm("UpdataDB-doTask", "Exception");
			e.printStackTrace();
		} finally {
		}
	}

	
}
